package com.study.flink;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.stream.Stream;

/**
 * @author xubangbang
 * @date 2021/7/6 12:28
 */
public class TestFlink {

    public static void main(String[] args) throws Exception {
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        String host = parameterTool.get("host");
        int port = parameterTool.getInt("port");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> data = env.socketTextStream(host, port);
        printStreamData(data);
        printStreamMax(data);

        env.execute("myJob");
    }

    private static void printStreamMax(DataStreamSource<String> data) {
        data.flatMap(new MaxFlatMapFunction()).print("当前流最大值");
    }

    private static void printStreamData(DataStreamSource<String> data) {
        data.flatMap(new FilterNumberFlatMapFunction()).print("当前数据流");
    }

    /**
     * 求 一条数据中最大的
     */
    public static class MaxFlatMapFunction implements FlatMapFunction<String, Long> {
        @Override
        public void flatMap(String s, Collector<Long> collector) {
            collector.collect(Stream.of(s.split(" ")).filter(e -> {
                try {
                    Long.parseLong(e);
                    return true;
                } catch (Exception e1) {
                    return false;
                }
            }).mapToLong(Long::parseLong).max().orElse(0));
        }
    }

    /**
     * 求 一条数据中最大的
     */
    public static class FilterNumberFlatMapFunction implements FlatMapFunction<String, String> {
        @Override
        public void flatMap(String s, Collector<String> collector) {
            StringBuilder sb = new StringBuilder();
            Stream.of(s.split(" ")).filter(e -> {
                try {
                    Long.parseLong(e);
                    return true;
                } catch (Exception e1) {
                    return false;
                }
            }).forEach(e -> sb.append(e).append(" "));
            collector.collect(sb.toString());
        }
    }
}
